package data_manipulate;


/*
payment table:
orderid userid payment productid
1,9819,100,121
2,8918,2000,111
3,2813,1234,22
4,9100,10,1101
5,3210,490,111
6,1298,28,1211
7,1010,281,90
8,1818,9000,20

we need to find top 5 payment
 */

import Utils.FileUtil;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

public class TopN {

    public static String inputPath = "hdfs://localhost:9000/top/top_payment";
    public static String outputPath = "hdfs://localhost:9000/top/output_payment";

    // in this sample, we extract all data in map process and hand them all to reduce process
    // for large datasets, this may case too much communication upkeep.
    // notice, we only need one reducer at this task!
    public static void main(String[] args)
            throws IOException, InterruptedException, ClassNotFoundException {

        Configuration conf = new Configuration();
        FileUtil fileUtil = new FileUtil(conf, outputPath);
        fileUtil.doError(inputPath);
        fileUtil.doDelete(outputPath);

        Job job = Job.getInstance(conf, "topN");
        job.setJarByClass(TopN.class);
        // for top problem, we only need on reducer process.
        job.setNumReduceTasks(1);

        job.setMapperClass(doMapper.class);
        job.setMapOutputKeyClass(MyIntWritable.class);
        job.setMapOutputValueClass(NullWritable.class);

        job.setReducerClass(doReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(MyIntWritable.class);

        job.setGroupingComparatorClass(doComparator.class);

        FileInputFormat.setInputPaths(job, new Path(inputPath));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }


    public static class MyIntWritable implements WritableComparable<MyIntWritable> {
        // we only store payment in this int object
        private int payment;

        public int getPayment() {
            return payment;
        }

        public void setPayment(int payment) {
            this.payment = payment;
        }

        public MyIntWritable() {
        }

        @Override
        public void write(DataOutput out) throws IOException {
            out.writeInt(payment);
        }

        @Override
        public void readFields(DataInput in) throws IOException {
            this.payment = in.readInt();
        }

        @Override
        public boolean equals(Object obj) {
            if (!(obj instanceof MyIntWritable)) {
                return false;
            }
            MyIntWritable o2 = (MyIntWritable) obj;
            return (this.payment == o2.payment);

        }

        @Override
        public String toString() {
            return payment + "";
        }

        @Override
        public int compareTo(MyIntWritable o) {
            return o.payment - this.payment;
        }
    }

    public static class doMapper extends Mapper<LongWritable, Text, MyIntWritable, NullWritable> {

        private MyIntWritable payment = new MyIntWritable();

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, MyIntWritable, NullWritable>.Context context) throws IOException, InterruptedException {
            String[] splits = value.toString().trim().split(",");
            // for safety check
            // a standard entry of data should be like:
            // orderid userid payment productid
            // we want payment, which is splits[2]
            if (splits.length == 4) {
                payment.setPayment(Integer.parseInt(splits[2]));
                context.write(payment, NullWritable.get());
                System.out.println(payment.toString());
            }
        }
    }

    public static class doReducer extends Reducer<MyIntWritable, NullWritable, Text, MyIntWritable> {

        private int idx = 0;
        private Text id = new Text();

        @Override
        protected void reduce(MyIntWritable key, Iterable<NullWritable> values, Reducer<MyIntWritable, NullWritable, Text, MyIntWritable>.Context context) throws IOException, InterruptedException {
            idx++;
            if (idx <= 5) {
                id.set(idx + "");
                System.out.println(key.toString());
                context.write(id, key);
            }
        }
    }

    public static class doComparator extends WritableComparator {

        protected doComparator() {
            super(MyIntWritable.class, true);
        }

        @Override
        public int compare(WritableComparable a, WritableComparable b) {
            MyIntWritable int1 = (MyIntWritable) a;
            MyIntWritable int2 = (MyIntWritable) b;
            return int1.compareTo(int2);
        }
    }

}
